package edu.csl.study.spark.basic

import org.apache.log4j.{Level, Logger}
import org.apache.spark.streaming._
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 用来替换updateStateByKey
 * 优势：更高的性能，功能更多
 */
object SparkStreaming_MapWithStateAPITest {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)

    val sparkConf = new SparkConf().setAppName("NetworkWordCount").setMaster("local[2]")
    val sc = new SparkContext(sparkConf)
    val ssc = new StreamingContext(sc, Seconds(2))

    ssc.checkpoint("/Users/lixiyuan/Desktop/data3")

    val lines = ssc.socketTextStream("192.168.100.20", 8090)

    val words = lines.flatMap(_.split(","))
    val wordsDStream = words.map(x => (x, 1))



    //初始时设置两个值
    val initialRDD = sc.parallelize(List(("hadoop", 100L), ("storm", 32L)))

    /**
     * hadoop,hadoop,hadoop
     * .....
     * hadoop,{1,1,1} =>
     */
    // currentBatchTime : 表示当前的Batch的时间
    // key: 表示需要更新状态的key
    // value: 表示当前batch的对应的key的对应的值  3
    // currentState: 对应key的上次的状态2
    val stateSpec =StateSpec.function((currentBatchTime: Time,
                                       key: String,
                                       value: Option[Int],
                                       currentState: State[Long]) => {

      //3 + 2 =5
      val sum = value.getOrElse(0).toLong + currentState.getOption.getOrElse(0L)
      //如果你的数据没有超时
      if (!currentState.isTimingOut()) {
        currentState.update(sum)
      }
      //最后一行代码是返回值
      val output = (key, sum)
      Some(output)
    }).initialState(initialRDD).timeout(Seconds(15))
    //timeout: 当一个key超过这个时间没有接收到数据的时候，这个key以及对应的状态会被移除掉

    /**
      * reduceByKey
      * udpateStateByKey
      * mapWithState
      */
    val result = wordsDStream.mapWithState(stateSpec)

   // result.print()  //只打印当前 更新的key的结果。
    result.stateSnapshots().print() //会打印所有key的结果
    //启动Streaming处理流
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }

}
